package br.ufc.great.dtx.tm.sch;

import java.io.IOException;
import java.io.ObjectInputStream;
import java.io.ObjectOutputStream;
import java.net.Socket;
import java.util.Vector;

import br.ufc.great.Graph;
import br.ufc.great.dtx.dm.DataManager;
import br.ufc.great.dtx.server.DTXOutput;
import br.ufc.great.dtx.server.config.Config;
import br.ufc.great.dtx.tm.AbortTransactionException;
import br.ufc.great.dtx.tm.CommitTransactionException;
import br.ufc.great.dtx.tm.Transaction;
import br.ufc.great.dtx.tm.deadlockdetector.DeadLockDaemon;
import br.ufc.great.dtx.tm.deadlockdetector.RemoteWFGMessage;
import br.ufc.great.dtx.tm.lm.LockIncompatibleException;
import br.ufc.great.dtx.tm.lm.LockManager;
import br.ufc.great.dtx.tm.operations.Operation;
import br.ufc.great.dtx.tm.operations.Site;
import br.ufc.great.xml.XMLDocument;

public class Scheduler extends Thread {

	// Gerenciador de Bloqueios
	private LockManager lockManager;
	// Gerenciador de Dados
	private DataManager dataManager;
	// Stream de Saída do Gerenciador de Transações
	private DTXOutput output;
	// Estrutura verificadora de Deadlock local
	private Graph waitFor;
	// Fila de transações que irão executar
	private Vector<Transaction> transactions;
	// Flag utilizado para finalizar a Thread Scheduler
	private boolean ended ;
	// Flag utilizado para indicar se o escalonador está em execução
	private boolean running;
	
	private DeadLockDaemon daemon;
	
	private SchedulerRemoteMessage schedulerRemoteMessage;
	private SchedulerRemoteOperation schedulerRemoteOperation;
	private SchedulerRemoteWFGMessage schedulerRemoteWFGMessage;
	
	/**
	 * Construtor que inicializa os recursos do escalonador
	 * 
	 * @param output
	 * @param dataManager
	 */
	public Scheduler(DTXOutput output, DataManager dataManager) {
		running = false;
		ended = false;
		this.output = output;
		this.dataManager = dataManager;
		XMLDocument document = null;
		synchronized (dataManager) {
			document = dataManager.getXMLDocument();
		}
		lockManager = new LockManager(output, document);
		output.print(getClass().getSimpleName(), "LockManager is started", DTXOutput.MESSAGE);
		waitFor = new Graph();
		transactions = new Vector<Transaction>();
		
		schedulerRemoteOperation = new SchedulerRemoteOperation(output, this, lockManager);
		schedulerRemoteMessage = new SchedulerRemoteMessage(output, this, lockManager, schedulerRemoteOperation);
		schedulerRemoteWFGMessage = new SchedulerRemoteWFGMessage(output, this);
		schedulerRemoteOperation.start();
		schedulerRemoteMessage.start();
		schedulerRemoteWFGMessage.start();
		
		daemon = new DeadLockDaemon(output, this);
		daemon.start();
		
		setPriority(MIN_PRIORITY);
		
		AlertDaemon ad = new AlertDaemon(output, this, schedulerRemoteOperation, schedulerRemoteMessage, schedulerRemoteWFGMessage);
		ad.start();
	}

	/**
	 * Método de execução do escalonador
	 */
	public void run() {
		output.print(getClass().getSimpleName(), "Scheduler is running", DTXOutput.MESSAGE);
		running = true;
		int i = 0;
		while (isAlive() && ! ended) {
			// Processa transações locais
			while (transactions.size() > 0) {
				Transaction transaction = null;
				synchronized (transactions) {
					if (i < transactions.size())
						transaction = transactions.elementAt(i);
					else {
						i = 0;
						continue;
					}
				}
				if (transaction == null) {
					i = 0;
					continue;
				}
				if (transaction.isAbort()) {
					abort(transaction.getId());
					i--;
				}
				else {
					try {
						Operation operation = transaction.nextOperation();
						Vector<Site> sites = operation.getSites();
						for (int j = 0; j < sites.size(); j++) {
							Site site = sites.elementAt(j);
							// Caso a operação seja remota
							if (! site.getHost().equalsIgnoreCase(Config.DTX_SERVER_HOST) || site.getPort() != Config.DTX_SERVER_PORT) {
								try {
									output.print(getClass().getSimpleName(), "Sending remote operation " + operation.getId() + " of transaction " + operation.getTransactionId() + " to " + site.getHost() + ":" + site.getPort(), DTXOutput.TEST);
									Socket socket = new Socket(site.getHost(), site.getPort());
									ObjectOutputStream out = new ObjectOutputStream(socket.getOutputStream());
									operation.setSites(new Vector<Site>());
									out.writeObject(operation);
									out.flush();
									output.print(getClass().getSimpleName(), "Operation " + operation.getId() + " of transaction " + operation.getTransactionId() + " was sent to " + site.getHost() + ":" + site.getPort(), DTXOutput.TEST);
									ObjectInputStream in = new ObjectInputStream(socket.getInputStream());								
									Operation remoteOperation = (Operation) in.readObject();
									operation.setSites(sites);
									switch (remoteOperation.getState()) {
										case Operation.STATE_EXECUTED : {
											operation.setState(remoteOperation.getState());
											operation.setResult(remoteOperation.getResult());
											output.print(getClass().getSimpleName(), "Operation " + operation.toString() + " " + "executed", DTXOutput.TEST);
											break;
										}
										case Operation.STATE_ABORTED : {
											transaction.setAbort(true);
											break;
										}
									}
									out.close();
									in.close();
									socket.close();
								}
								catch (IOException e) {
									output.print(getClass().getSimpleName(), e.getMessage(), DTXOutput.ERROR);
								}
								catch (ClassNotFoundException e) {
									output.print(getClass().getSimpleName(), e.getMessage(), DTXOutput.ERROR);
								}
							}
							// Caso a opera��o seja local
							else {
								processOperation(transaction, operation);
							}
							if (operation.getState() == Operation.STATE_EXECUTED && (j < sites.size() - 1))
								operation.setState(Operation.STATE_NONE);
						}
					}
					catch (AbortTransactionException e) {
						abort(transaction.getId());
						i--;
					}
					catch (CommitTransactionException e) {
						commit(transaction.getId());
						i--;
					}
				}
				i++;
				if (i == -1 || i == transactions.size())
					i = 0;
			}
			if (transactions.size() > 0)
				continue;
			pause();
		}
	}

	/**
	 * M�todo utilizado para colocar a thread escalonador no estado de espera
	 */
	public synchronized void pause() {
		try {
			output.print(getClass().getSimpleName(), "is waiting", DTXOutput.TEST);
			wait();
		}
		catch (InterruptedException e) {
			e.printStackTrace();
		}
	}

	/**
	 * M�todo utilizado para reiniciar a thread escalonador
	 */
	public synchronized void restart() {
		notify();
	}
	
	/**
	 * M�todo utilizado para adicionar uma transa��o na fila de transa��es
	 * 
	 * @param transaction
	 */
	public void sendTransaction(Transaction transaction) {
		transactions.add(transaction);
		output.print(getClass().getSimpleName(), "Transaction " + transaction.getId() + " was received", DTXOutput.TEST);
		restart();
	}
	
	/**
	 * M�todo utilizado para adicionar uma opera��o remota a ser executada por este escalonador
	 * 
	 * @param remoteOperation
	 */
	public void sendRemoteOperation(RemoteOperation remoteOperation) {
		schedulerRemoteOperation.addRemoteOperation(remoteOperation);
	}

	public void sendRemoteMessage(RemoteMessage remoteMessage) {
		schedulerRemoteMessage.addRemoteMessage(remoteMessage);
	}

	public void sendRemoteWFGMessage(RemoteWFGMessage remoteWFGMessage) {
		schedulerRemoteWFGMessage.addRemoteWGFMessage(remoteWFGMessage);
	}

	/**
	 * M�todo utilizado pelo escalonador para abortar uma transa��o
	 * 
	 * @param transaction
	 */
	public void abort(String transactionId) {
		ProcessAbort process = new ProcessAbort(output, this, lockManager, transactionId);
		process.start();
	}
	
	/**
	 * M�todo utilizado pelo escalonador para consolidar uma transa��o
	 * 
	 * @param transaction
	 */
	public void commit(String transactionId) {
		ProcessCommit process = new ProcessCommit(output, this, lockManager, transactionId);
		process.start();
	}
	
	/**
	 * M�todo utilizado para processar uma consulta no documento pelo DataGuide
	 * 
	 * @param queryOperation
	 * @throws AbortTransactionException 
	 */
	private void processOperation(Transaction transaction, Operation operation) throws AbortTransactionException {
		String result = null;
		try {
			output.print(getClass().getSimpleName(), "Operation " + operation.toString() + " " + "is processing", DTXOutput.TEST);
			synchronized (lockManager) {
				result = lockManager.processOperation(operation);
			}
			output.print(getClass().getSimpleName(), "Operation " + operation.toString() + " " + "executed", DTXOutput.TEST);
			operation.setResult(result);
			operation.setState(Operation.STATE_EXECUTED);
		}
		catch (LockIncompatibleException e) {
			synchronized (lockManager) {
				lockManager.unlock(transaction.getId(), operation.getId());
			}
			processLockIncompatible(transaction, operation, e);
			output.print(getClass().getSimpleName(), "Operation " + operation.toString() + " " + "not processed", DTXOutput.TEST);
		}
	}
	
	/**
	 * M�todo utilizado para processar o n�o concedimento de um bloqueio para a opera��o. Caso ocorra deadlock a exce��o de aborto ser� lan�ada, se n�o ocorrer deadlock mas o bloqueio n�o for concedido, a transa��o ser� posta em espera.
	 * 
	 * @param operation
	 * @param exception
	 */
	private void processLockIncompatible(Transaction transaction, Operation operation, LockIncompatibleException exception) throws AbortTransactionException {
		waitFor.addLink(operation.getTransactionId(), exception.getTransactionIdIncompatible());
		if (waitFor.isCircle(operation.getTransactionId())) {
			output.print(getClass().getSimpleName(), "Deadlock find between " + operation.getTransactionId() + " and " + exception.getTransactionIdIncompatible() + ", transaction " + operation.getTransactionId() + " will aborted", DTXOutput.TEST);
			throw new AbortTransactionException();
		}
		else {
			output.print(getClass().getSimpleName(), "Transaction " + transaction.getId() + " is waiting for locking...", DTXOutput.TEST);
		}
	}
	
	/**
	 * M�todo utilizado para encerrar o escalonador
	 */
	public void end() {
		ended = true;
		restart();
	}
	
	/**
	 * M�todo utilizado para verificar se o escalonador est� em execu��o
	 * 
	 * @return
	 */
	public boolean isRunning() {
		return running;
	}
	
	public void removeTransaction(Transaction transaction) {
		synchronized (transactions) {
			for (int i = 0; i < transactions.size(); i++) {
				Transaction transactionItem = transactions.elementAt(i);
				if (transactionItem.getId().equalsIgnoreCase(transaction.getId())) {
					transactions.remove(i);
					return;
				}
			}
		}
	}
	
	public Transaction getTransaction(String id) {
		synchronized (transactions) {
			for (int i = 0; i < transactions.size(); i++) {
				Transaction transactionItem = transactions.elementAt(i);
				if (transactionItem.getId().equalsIgnoreCase(id)) {
					return transactionItem;
				}
			}
			return null;
		}
	}
	
	public int transactionsSize() {
		synchronized (transactions) {
			return transactions.size();
		}
	}
	
	public Graph getWaitForGraph() {
		synchronized (waitFor) {
			return waitFor;
		}
	}

	public synchronized int size() {
		return transactions.size();
	}

}
